package org.caw.hbase.admin.impl;

import java.io.IOException;

import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.HConnectionManager;
import org.apache.hadoop.hbase.util.Bytes;
import org.caw.domain.descriptor.TableDescriptor;
import org.caw.domain.descriptor.TableDescriptorList;
import org.caw.exception.CawAdminException;
import org.caw.exception.CawException;
import org.caw.exception.CawNoTableFoundException;
import org.caw.exception.CawTableAlreadyExistsException;
import org.caw.hbase.HBaseConfigurationBean;
import org.caw.hbase.admin.AdminManager;
import org.caw.util.ConvertUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class HBaseAdminManagerImpl implements AdminManager {

	private static final Logger LOGGER = LoggerFactory.getLogger(HBaseAdminManagerImpl.class);

	private HBaseConfigurationBean configuration;

	private CachedHBaseAdmin admin;

	public void init() throws MasterNotRunningException, ZooKeeperConnectionException {
		admin = new CachedHBaseAdmin(configuration.getConf());
	}

	public void destroy() {
		HConnectionManager.deleteAllConnections(true);
	}

	@Override
	public TableDescriptorList getTableDescriptions(Boolean detailed) throws CawException {
		try {
			TableDescriptorList result = new TableDescriptorList();
			if (Boolean.FALSE.equals(detailed)) {
				for (TableDescriptor desc : admin.listTablesFromCache()) {
					desc.setFamilyDescriptors(null);
					result.getTables().add(desc);
				}
			} else {
				result.getTables().addAll(admin.listTablesFromCache());
			}
			LOGGER.info("Successfully listed all HBase tables");
			return result;
		} catch (IOException e) {
			LOGGER.error("Failed to list HBase tables", e);
			throw new CawException("Failed to list HBase tables", e);
		}
	}

	@Override
	public TableDescriptor getTableDescription(String tableName) throws CawException {
		try {
			if (!admin.tableExists(tableName)) {
				throw new CawNoTableFoundException("Table " + tableName + " doesn't exist");
			}
			HTableDescriptor desc = admin.getTableDescriptor(Bytes.toBytes(tableName));
			LOGGER.info("Successfully get descriptor for table {}", tableName);
			TableDescriptor d = ConvertUtil.convert(desc, Boolean.TRUE);
			d.setEnabled(admin.isTableEnabled(tableName));
			return d;
		} catch (IOException e) {
			LOGGER.error("Failed to get descriptor for table {}", tableName, e);
			throw new CawException("Failed to get descriptor for table " + tableName, e);
		}
	}

	@Override
	public TableDescriptor createTable(TableDescriptor descriptor, boolean overwrite) throws CawException {
		try {
			if (admin.tableExists(descriptor.getName())) {
				if (overwrite) {
					deleteTable(descriptor.getName());
				} else {
					throw new CawTableAlreadyExistsException("Table " + descriptor.getName() + " already exist");
				}
			}
			HTableDescriptor desc = ConvertUtil.convert(descriptor);
			admin.createTable(desc);
			LOGGER.info("Table {} successfully created", descriptor.getName());
			return getTableDescription(descriptor.getName());
		} catch (IOException e) {
			LOGGER.error("Failed to create table table {}", descriptor.getName(), e);
			throw new CawException("Failed to create table table " + descriptor.getName(), e);
		}
	}

	@Override
	public TableDescriptor updateTable(TableDescriptor descriptor) throws CawException {

		// FIXME in cases of errors keep valid state on cloud
		try {
			if (!admin.tableExists(descriptor.getName())) {
				throw new CawNoTableFoundException("Table " + descriptor.getName() + " doesn't exist");
			}
			HTableDescriptor deltaDescriptor = ConvertUtil.convert(descriptor);

			admin.disableTable(descriptor.getName());
			for (HColumnDescriptor colDesc : deltaDescriptor.getFamilies()) {
				admin.addColumn(descriptor.getName(), colDesc);
			}
			admin.enableTable(descriptor.getName());

			return ConvertUtil.convert(admin.getTableDescriptor(Bytes.toBytes(descriptor.getName())), Boolean.TRUE);
		} catch (IOException e) {
			throw new CawAdminException("Failed to update table. Reason: " + e.getMessage(), e);
		}
	}

	@Override
	public void disableTable(String tableName) throws CawException {
		try {
			if (!admin.tableExists(tableName)) {
				throw new CawNoTableFoundException("Table " + tableName + " doesn't exist");
			}
			admin.disableTable(tableName);
		} catch (IOException e) {
			throw new CawAdminException("Failed to disable table", e);
		}
	}

	@Override
	public void enableTable(String tableName) throws CawException {
		try {
			if (!admin.tableExists(tableName)) {
				throw new CawNoTableFoundException("Table " + tableName + " doesn't exist");
			}
			admin.enableTable(tableName);
		} catch (IOException e) {
			throw new CawAdminException("Failed to enable table", e);
		}
	}

	@Override
	public void trunateTable(String tableName) throws CawException {
		try {
			// FIXME better exception handling to preserve valid state on cloud
			if (!admin.tableExists(tableName)) {
				throw new CawNoTableFoundException("Table " + tableName + " doesn't exist");
			}
			HTableDescriptor desc = admin.getTableDescriptor(Bytes.toBytes(tableName));
			admin.disableTable(tableName);
			admin.deleteTable(tableName);
			admin.createTable(desc);
		} catch (IOException e) {
			throw new CawAdminException("Failed to truncate table", e);
		}
	}

	@Override
	public void deleteTable(String tableName) throws CawException {
		try {
			if (!admin.tableExists(tableName)) {
				throw new CawNoTableFoundException("Table " + tableName + " doesn't exist");
			}
			disableTable(tableName);
			admin.deleteTable(tableName);
		} catch (IOException e) {
			throw new CawAdminException("Failed to delete table", e);
		}
	}

	public HBaseConfigurationBean getConfiguration() {
		return configuration;
	}

	public void setConfiguration(HBaseConfigurationBean configuration) {
		this.configuration = configuration;
	}

}